package org.example.realtime.traffic.dwd.log.function;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.realtime.jtp.common.utils.DateTimeUtil;
import org.example.realtime.traffic.dwd.log.entity.TrafficFlowType;

/**
 * @Title: FlowTypeReportFunction
 * @Author Lianzy
 * @Package org.example.realtime.traffic.dwd.log.function
 * @Date 2025/5/29 18:10
 * @description
 */
public class FlowTypeReportFunction implements WindowFunction<TrafficFlowType, String, String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow window, Iterable<TrafficFlowType> input, Collector<String> out) throws Exception {

        // 获取窗口开始和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // 对窗口中数据进行计算
        Long VehicleTypeCount = 0L;
        for (TrafficFlowType pageViewBean : input) {
            VehicleTypeCount += pageViewBean.getVehicleTypeCount();
        }

        // 拼接字符串
        String output = windowStartTime + "," +
                windowEndTime +"," +
                s + "," +
                VehicleTypeCount + "," +
                System.currentTimeMillis();

        out.collect(output);
//        System.out.println("输出结果 : " + output);


    }
}